package demo.kafka.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

@Configuration
public class ConsumerCfg {

    private final AtomicInteger messageCounter = new AtomicInteger(0);

    /**
     * 消息消费者 - 消费处理后的消息
     * 这是一个Consumer函数，只消费不产生输出
     */
    @Bean
    public Consumer<TextProcessorCfg.ProcessedMessageEvent> messageConsumer() {
        return message -> {
            int count = messageCounter.incrementAndGet();
            System.out.println("=========================================");
            System.out.println("📥 Consumed Message #" + count);
            System.out.println("Message ID: " + message.getMessageId());
            System.out.println("Original: " + message.getOriginalContent());
            System.out.println("Processed: " + message.getProcessedContent());
            System.out.println("Word Count: " + message.getWordCount());
            System.out.println("Sentiment: " + message.getSentiment());
            System.out.println("Contains Kafka: " + message.isContainsKafka());
            System.out.println("Processed At: " + message.getProcessedAt());
            System.out.println("Status: " + message.getStatus());
            System.out.println("=========================================\n");

            // 这里可以添加业务逻辑，比如保存到数据库、发送通知等
            // saveToDatabase(message);
            // sendNotification(message);
        };
    }

    public int getMessageCount() {
        return messageCounter.get();
    }
}
